/**
 * @FileName:NettyClientRpcHandler
 * @Package: com.std.netty.rpc
 * @author sence
 * @created 4/14/2015 11:30 AM
 * <p>
 * Copyright 2011-2015 Asura
 */
package com.ziroom.sms.std.rpc;

import com.ziroom.sms.std.rpc.api.Invocation;
import com.ziroom.sms.std.rpc.exception.RpcException;
import com.ziroom.sms.std.rpc.protocol.RpcResponse;
import io.netty.channel.*;

import java.util.concurrent.ConcurrentHashMap;

/**
 *
 * <p>Netty Client</p>
 *
 * <PRE>
 * <BR>	修改记录
 * <BR>-----------------------------------------------
 * <BR>	修改日期			修改人			修改内容
 * </PRE>
 *
 * @author sence
 * @since 1.0
 * @version 1.0
 */
public class NettyClientRpcHandler extends ChannelInboundHandlerAdapter {

    private volatile Channel channel;
    //这儿有一个并发的问题 生产出来的结果不是消费者想要的
//	private final BlockingQueue<Object> answer = new LinkedBlockingQueue<Object>();
    private ConcurrentHashMap<Object, Object> map = new ConcurrentHashMap<>();

    public Object sendRpcInvoke(Invocation invocation) throws RpcException {
        channel.writeAndFlush(new CustomInvocation());
        Object resObj ;
        for (; ; ) {
            if (map.containsKey(invocation.getToken())) {
                resObj = map.get(invocation.getToken());
                break;
            }
        }
        return resObj;

    }

    /**
     * Calls {@link io.netty.channel.ChannelHandlerContext#fireChannelRegistered()} to forward
     * to the next {@link ChannelHandler} in the {@link ChannelPipeline}.
     *
     * Sub-classes may override this method to change behavior.

     * @param ctx
     */
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        channel = ctx.channel();
    }

    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof RpcResponse) {
            RpcResponse rr = (RpcResponse) msg;
            map.put(rr.getToken(), rr.getResult());
        } else {
            throw new RpcException("rpc failed");
        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}
